package com.qyer.lxp.common.framework.threadpool;

import org.apache.log4j.Logger;

import java.io.IOException;
import java.io.InputStream;
import java.text.SimpleDateFormat;
import java.util.Collection;
import java.util.Date;
import java.util.Properties;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 公共线程池，用来执行异步任务<br/>
 * 线程池的相关信息在配置文件中设置:<br/>
 * 首先会从ClassPath根路径下寻找配置文件，找不到将使用默认配置文件
 * 默认配置文件在当前包的resource目录下
 *
 * @author jiangping
 * @since 1.0
 */

public final class CommonThreadPool {

    private static final Logger logger =
            Logger.getLogger(CommonThreadPool.class);
    private static ExecutorService execute = init();
    public static final String LONG_FORMAT = "yyyy-MM-dd HH:mm:ss";

    private CommonThreadPool() {
    }

    /**
     * 添加异步任务
     *
     * @param command
     * @return future, 返回任务等待对象
     */
    public static <T> Future<T> execute(IAroundAdviceCallable command) {
        Future<T> future = execute.submit(command);
        return future;
    }

    /**
     * 关闭线程池
     * @see ThreadPoolExecutor
     *
     * @return
     */
    public static boolean shutDown() {
        if (execute != null) {
            execute.shutdown();
            return true;
        }
        return false;
    }

    /**
     * 获取线程池对象
     *
     * @param vo
     * @return
     */
    private static RunningCounterThreadPoolExecutor getThreadPool(
            ThreadPoolParameterVO vo) {

        int corePoolSize = vo.getCorePoolSize();
        int maximumPoolSize = vo.getMaximumPoolSize();
        int initialCapacity = vo.getInitialCapacity();
        long keepAliveTime = vo.getKeepAliveTime();
        String threadName = vo.getThreadName();

        //增加构造队列容量参数
        TaskQueue taskqueue = new TaskQueue(initialCapacity);
        RunningCounterThreadPoolExecutor executeNew =
                new RunningCounterThreadPoolExecutor(
                        corePoolSize, maximumPoolSize,
                        keepAliveTime, TimeUnit.SECONDS,
                        taskqueue, new TaskThreadFactory(threadName),
                        new ThreadPoolRejectedExecutionHandler());
        taskqueue.setParent(executeNew);
        return executeNew;
    }

    private static ExecutorService init() {
        Properties ps = getThreadPoolConfig();
        int corePoolSize = Integer.parseInt(ps.getProperty("corePoolSize", "5"));
        int maximumPoolSize = Integer.parseInt(ps.getProperty("maximumPoolSize", "120"));
        int initialCapacity = Integer.parseInt(ps.getProperty("initialCapacity", "20000"));
        long keepAliveTime = Long.parseLong(ps.getProperty("keepAliveTime", "120"));
        String threadName = ps.getProperty("threadName", "common-threadPool-");
        ThreadPoolParameterVO vo = new ThreadPoolParameterVO();
        vo.setCorePoolSize(corePoolSize);
        vo.setMaximumPoolSize(maximumPoolSize);
        vo.setInitialCapacity(initialCapacity);
        vo.setKeepAliveTime(keepAliveTime);
        vo.setMaximumPoolSize(maximumPoolSize);
        vo.setThreadName(threadName);
        return getThreadPool(vo);
    }


    private static Properties getThreadPoolConfig() {
        Properties ps = new Properties();
        //先从ClassPath根路径找，找不到就使用当前类目录下找
        InputStream in =
                CommonThreadPool.class.getResourceAsStream(
                        "/threadPoolConfig.properties");
        if (in == null) {
            logger.warn("ClassPath根路径找不到 threadpool 配置文件，将使用默认配置文件");
            in =
                    CommonThreadPool.class.getResourceAsStream(
                            "/threadPoolConfig.properties");
            if (in == null) {
                logger.error("threadpool 默认配置文件不存在");
                throw new NullPointerException("threadpool 默认配置文件不存在");
            }
        }
        try {
            ps.load(in);
        } catch (IOException e) {
            throw new RuntimeException("读取threadpool 配置文件异常", e);
        } finally {
            if (in != null) {
                try {
                    in.close();
                } catch (IOException e) {
                    logger.error("关闭threadpool 配置文件输入流异常\", e");
                }
            }
        }
        return ps;
    }

    /**
     * 线程工厂
     */
    static class TaskThreadFactory implements ThreadFactory {
        final ThreadGroup group;
        final AtomicInteger threadNumber = new AtomicInteger(1);
        final String namePrefix;

        TaskThreadFactory(String namePrefix) {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
            this.namePrefix = namePrefix;
        }

        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement());
            t.setDaemon(true);
            if (t.getPriority() != Thread.NORM_PRIORITY) {
                t.setPriority(Thread.NORM_PRIORITY);
            }
            return t;
        }
    }

    /**
     * 当线程池的线程数不小于corePoolSize时，默认情况下
     * {@link ThreadPoolExecutor}
     * 会调用任务队列的offer方法来加入任务<br/>
     * 若返回false（队列已满）则创建临时线程（闲置一段时间会被回收）直至线程池已满<br/>
     * 本队列会重构offer方法，若待处理的任务数大于线程池现有线程数且线程池未满也返回false，
     * 保证不需要等到任务队列满了才创建临时线程<br/>
     * <pre>
     * TODO 也可以采用覆盖ThreadPoolExecutor的execute方法来实现，后续再优化
     * </pre>
     */
    static class TaskQueue extends LinkedBlockingQueue<Runnable> {

        private static final long serialVersionUID = -3966913824895982184L;
        RunningCounterThreadPoolExecutor parent = null;

        public TaskQueue() {
            super();
        }

        public TaskQueue(int initialCapacity) {
            super(initialCapacity);
        }

        public TaskQueue(Collection<? extends Runnable> c) {
            super(c);
        }

        public void setParent(RunningCounterThreadPoolExecutor tp) {
            parent = tp;
        }

        public boolean force(Runnable o) {
            if (parent.isShutdown()) {
                throw new RejectedExecutionException("Executor not running, can't force a command into the queue");
            }
            return super.offer(o); //forces the item onto the queue, to be used if the task is rejected
        }

        public boolean offer(Runnable o) {
            //we can't do any checks
            if (parent == null) {
                return super.offer(o);
            }
            //we are maxed out on threads, simply queue the object
            if (parent.getPoolSize() == parent.getMaximumPoolSize()) {
                return super.offer(o);
            }
            //we have idle threads, just add it to the queue
            //note that we don't use getActiveCount()
            AtomicInteger submittedTasksCountNew = parent.uncompletedTasksCount;
            if (submittedTasksCountNew != null
                    && submittedTasksCountNew.get() <= parent.getPoolSize()) {
                return super.offer(o);
            }
            //if we have less threads than maximum force creation of a new thread
            if (parent.getPoolSize() < parent.getMaximumPoolSize()) {
                return false;
            }
            //if we reached here, we need to add it to the queue
            return super.offer(o);
        }
    }

    /**
     * 自定义线程池任务终止实现
     */
    static class ThreadPoolRejectedExecutionHandler implements RejectedExecutionHandler {
        private final Logger logger = Logger.getLogger(this.getClass());

        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            //再次尝试加入任务队列（针对并发情况导致的失败)
            try {
                boolean reAdd = executor.getQueue().offer(r, 10, TimeUnit.MILLISECONDS);
                if (reAdd) {
                    return;
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //重试失败后，记录日志
            StringBuilder sb = new StringBuilder();
            if (r instanceof AroundAdviceFutureTask) {
                IAroundAdviceCallable aroundAdviceCallable =
                        ((AroundAdviceFutureTask) r).getArounAdviceCallable();
                sb.append("任务名称:").append(aroundAdviceCallable.getClass());
                sb.append(", happenTime=").append(formateDate());
                sb.append(",toString=").append(aroundAdviceCallable.toString());
            } else {
                sb.append("任务名称:").append(r.getClass());
                sb.append(", happenTime=").append(formateDate());
                sb.append(",toString=").append(r.toString());
            }
            logger.warn("CommonThreadPool 以达到队列容量上限：" + sb.toString());

            //正在执行任务计数器减1
            if (executor instanceof RunningCounterThreadPoolExecutor) {
                ((RunningCounterThreadPoolExecutor) executor).getUncompletedTasksCount().decrementAndGet();
            }
            throw new RejectedExecutionException();
        }

        private String formateDate() {
            Date date = new Date();
            SimpleDateFormat sdf = new SimpleDateFormat(LONG_FORMAT);
            String result = sdf.format(date);
            return result;
        }

    }

}
